package com.hung.triple.transport.netty4;


import com.hung.triple.common.exception.TripleRpcException;
import com.hung.triple.common.utils.AddressUtils;
import com.hung.triple.protocol.TripleProtocol;
import com.hung.triple.transport.Transporter;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.AllArgsConstructor;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 数据发送器
 *
 * @author Hung
 * @since 2022/9/18
 */
@AllArgsConstructor
public class NettyTransporter implements Transporter {

    // Netty 数据通道
    private final Channel channel;

    //权重
    private int weight = 100;

    private ReentrantLock lock = new ReentrantLock();


    public NettyTransporter(Channel channel) {
        this.channel = channel;
    }

    @Override
    public void promoteWeight() {
        lock.lock();
        try {
            if (weight == 100) {
                return;
            }
            weight *= 2;
        } finally {
            lock.unlock();
        }

    }

    @Override
    public void lowerWeight() {
        lock.lock();
        try {
            if (weight == 1) {
                weight = 0;
            }
            weight /= 2;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public boolean available() {
        return channel.isActive();
    }

    @Override
    public void sendSync(TripleProtocol protocol, long timeoutMS) throws TripleRpcException {
        ChannelFuture future = channel.writeAndFlush(protocol);
        try {
            future.get(timeoutMS, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            processFailed(protocol);
            throw new TripleRpcException("time out", e);
        }
    }

    @Override
    public void sendAsync(TripleProtocol protocol) {
        ChannelFuture channelFuture = channel.writeAndFlush(protocol);
        channelFuture.addListener(f -> {
            if (!f.isSuccess()) {
                processFailed(protocol);
            }
        });
    }

    @Override
    public String remoteAddress() {
        return AddressUtils.getAddress(channel);
    }

    @Override
    public Integer getWeight() {
        return weight;
    }

    @Override
    public Channel getChannel() {
        return channel;
    }
}
